package com.zhm.demo.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

@Slf4j
public class CanalTest {
    public static void main(String[] args) {
        System.out.println("start main");
        new CanalTest().initThread();
    }


    private Map<String, String> errorMap = new HashMap<>();


    private void initThread() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        initConnect();
                    } catch (Exception e) {
                        String key = "canal_connection_error";
                        if (!hasSameError(key, e.getMessage())) {
                            log.error("canal连接出错: {}", e);
                        }
                    }
                    try {
                        Thread.sleep(10000);
                    } catch (InterruptedException e) {
                    }
                }
            }
        }).start();
    }

    private void initConnect() {


        String canalMonitorHost = "127.0.0.1";

        Integer canalMonitorPort = 11111;

        String userName = "";

        String password = "";

        String canalIp = canalMonitorHost;
        int canalPort = canalMonitorPort;
        String canalDestination = "example";
        String canalUsername = userName;
        String canalPassword = password;
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalIp,
                canalPort), canalDestination, canalUsername, canalPassword);
        int batchSize = 200;
        try {
            connector.connect(); // 连接到canal server
            connector.subscribe(".*\\..*"); // 订阅指定的消息
            connector.rollback(); // 回滚到未进行ack 的地方
            log.info("canal连接成功");
            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        //未获取到消息则睡眠
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    try {
                        //处理消息
                        log.info("从canal接收到: {} 条消息,消息批次: {}，开始处理", size, message.getId());
                        handleMessage(message.getEntries());
                    } catch (Exception e) {
                        connector.rollback(batchId); // 处理失败, 回滚数据
                        String key = "canal_sync_data_error";
                        String errMsg = e.getMessage();
                        if (StringUtils.isEmpty(errMsg)) errMsg = e.toString();
                        if (!hasSameError(key, errMsg)) {
                            log.error("同步数据出错: {}", e);
                        }
                        //休眠一段时间继续获取数据
                        try {
                            Thread.sleep(10000);
                        } catch (InterruptedException ex) {
                            ex.printStackTrace();
                        }
                        continue;
                    }
                }
                connector.ack(batchId); // 提交确认
            }
        } finally {
            connector.disconnect();
        }
    }

    private boolean hasSameError(String key, String error) {
        String lastError = errorMap.get(key);
        if (Objects.equals(lastError, error)) {
            return true;
        }
        errorMap.put(key, error);
        return false;
    }

    private void handleMessage(List<CanalEntry.Entry> entrys) throws InvalidProtocolBufferException {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            //根据数据库名获取租户名
            String databaseName = entry.getHeader().getSchemaName();
            String tableName = entry.getHeader().getTableName();
            log.info("数据库: {}, 表名: {}", databaseName, tableName);
            // 获取类型
            CanalEntry.EntryType entryType = entry.getEntryType();

            // 获取序列化后的数据
            ByteString storeValue = entry.getStoreValue();
            if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
                // 反序列化数据
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                // 获取当前事件的操作类型
                CanalEntry.EventType eventType = rowChange.getEventType();
                if (eventType == CanalEntry.EventType.INSERT || eventType == CanalEntry.EventType.UPDATE
                        || eventType == CanalEntry.EventType.DELETE) {
                    // 获取数据集
                    List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
                    // 遍历rowDataList，并打印数据集
                    for (CanalEntry.RowData rowData : rowDataList) {
                        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                        List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                        // 变更前数据
                        for (CanalEntry.Column column : beforeColumnsList) {
                            log.info("变更前数据: name: {}, value: {}", column.getName(), column.getValue());
                        }
                        // 变更后数据
                        for (CanalEntry.Column column : afterColumnsList) {
                            log.info("变更后数据: name: {}, value: {}", column.getName(), column.getValue());
                        }
                    }
                }
            }
        }
    }

}
